package com.demo.config.canal.test3;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.demo.config.canal.redisAndRabbitMQ.Constants;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;

/**
 * canal任务执行类： 说明：因不同的表存入缓存时可能设置的key不同以及缓存查策略不同，因此这里使用策略模式，不同的表同步缓存时单独处理，策略的key定义为表名，
 * 有新的表要实现同步缓存时，只需要新增一个策略就行，不用改动其他代码，保证代码良好的扩展性。
 */
@Slf4j
public class CanalMySQL2RedisTask implements Runnable {

	private String canalServerHost;

	private int canalServerPort;

	private String canalServerDesrination;

	private String dataBase;

	private int batchSizeInit;

	// 策略上下文
	private TableStrategyContext tableStrategyContext;

	public CanalMySQL2RedisTask(String canalServerHost, int canalServerPort, String canalServerDesrination,
			String dataBase, int batchSizeInit, TableStrategyContext tableStrategyContext) {
		this.canalServerHost = canalServerHost;
		this.canalServerPort = canalServerPort;
		this.canalServerDesrination = canalServerDesrination;
		this.dataBase = dataBase;
		this.batchSizeInit = batchSizeInit;
		this.tableStrategyContext = tableStrategyContext;
	}

	@Override
	public void run() {
		// 创建客户端
		CanalConnector connector = CanalConnectors.newSingleConnector(
				new InetSocketAddress(canalServerHost, canalServerPort), canalServerDesrination, Constants.BLANK_STR,
				Constants.BLANK_STR);
		try {
			// 批量个数
			int batchSize = batchSizeInit;
			// 连接
			connector.connect();
			// 过滤监听的库-表
			connector.subscribe(dataBase + ".*");
			while (true) {
				// 获取指定数量的数据
				Message message = connector.getWithoutAck(batchSize);
				long batchId = message.getId();
				int size = message.getEntries().size();
				// log.info("batchId = {}，size = {}", batchId, size);
				List<CanalEntry.Entry> entries = message.getEntries();
				if (CollectionUtils.isEmpty(entries)) {
					try {
						Thread.sleep(300);
					}
					catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				else {
					// 处理数据
					handleEntry(message.getEntries());
				}
				// 提交确认
				connector.ack(batchId);
			}
		}
		catch (Exception e) {
			log.error("exception e ," + e);
			// 异常回滚：回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
			connector.rollback();
		}
		finally {
			// 释放连接
			connector.disconnect();
		}
	}

	private void handleEntry(List<CanalEntry.Entry> entrys) {
		for (CanalEntry.Entry entry : entrys) {
			if (entry.getEntryType().equals(CanalEntry.EntryType.ROWDATA)) {
				CanalEntry.RowChange rowChage = null;
				try {
					rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
				}
				catch (Exception e) {
					throw new RuntimeException(
							"ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
				}
				CanalEntry.EventType eventType = rowChage.getEventType();
				String tableName = entry.getHeader().getTableName();
				log.info("tableName ：{}, eventType : {}", tableName, eventType);

				/** 后续通过表名 对不同的表数据分别进行处理 **/

				for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
					if (eventType == CanalEntry.EventType.DELETE) {
						redisDelete(rowData.getBeforeColumnsList(), tableName);
					}
					else if (eventType == CanalEntry.EventType.INSERT) {
						redisInsert(rowData.getAfterColumnsList(), tableName);
					}
					else {
						log.info("-------> before");
						printColumn(rowData.getBeforeColumnsList());
						log.info("-------> after");
						redisUpdate(rowData.getAfterColumnsList(), tableName);
					}
				}
			}
		}
	}

	private void printColumn(List<CanalEntry.Column> columns) {
		for (CanalEntry.Column column : columns) {
			log.info(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
		}
	}

	private void redisInsert(List<CanalEntry.Column> columns, String tableName) {
		JSONObject json = new JSONObject();
		for (CanalEntry.Column column : columns) {
			json.set(FieldConvertUtil.lineToHump(column.getName()), column.getValue());
		}
		if (columns.size() > 0) {
			// 通过表名获取策略
			TableOperationStrategy resource = tableStrategyContext.getResource(tableName);
			if (resource != null) {
				resource.insert(JSONUtil.toJsonStr(json));
			}
		}
	}

	private void redisUpdate(List<CanalEntry.Column> columns, String tableName) {
		JSONObject json = new JSONObject();
		for (CanalEntry.Column column : columns) {
			json.set(FieldConvertUtil.lineToHump(column.getName()), column.getValue());
		}
		if (columns.size() > 0) {
			// 通过表名获取策略
			TableOperationStrategy resource = tableStrategyContext.getResource(tableName);
			if (resource != null) {
				resource.update(JSONUtil.toJsonStr(json));
			}
		}
	}

	private void redisDelete(List<CanalEntry.Column> columns, String tableName) {
		JSONObject json = new JSONObject();
		for (CanalEntry.Column column : columns) {
			json.set(FieldConvertUtil.lineToHump(column.getName()), column.getValue());
		}
		if (columns.size() > 0) {
			// 通过表名获取策略
			TableOperationStrategy resource = tableStrategyContext.getResource(tableName);
			if (resource != null) {
				resource.delete(JSONUtil.toJsonStr(json));
			}
		}
	}

}
